#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__

#include "../mqcommon/logger.hpp"

#include "../mqcommon/uuider.hpp"

#include "../mqcommon/thread_pool.hpp"
#include "../mqcommon/message.pb.h"
#include "../mqcommon/protocol.pb.h"

#include "../third/include/muduo/net/TcpConnection.h"
#include "../third/include/muduo/proto/codec.h"
#include "../third/include/muduo/proto/dispatcher.h"

#include "consumer.hpp"

#include <mutex>
#include <condition_variable>
#include <iostream>
#include <unordered_map>

namespace zhong
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicResponsePtr = std::shared_ptr<basicResponse>;
    using basicUserResponsePtr = std::shared_ptr<basicUserResponse>;
    using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;

        Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
            : _channel_id(UUIDer::uuid()), _connection(conn), _codec(codec) {}
        ~Channel()
        {
            // 信道关闭时需要取消订阅
            basicCancel();
            _basic_response_map.clear();
        }

        /* 提供给用户的接口函数，当发送了请求之后，需要获得响应才能够返回！！！即结束当前函数模块*/
        bool openChannel()
        {
            std::string request_id = UUIDer::uuid();
            // 给服务端发送信道创建请求
            openChannelRequest request;

            request.set_channel_id(_channel_id);
            request.set_request_id(request_id);
            _codec->send(_connection, request);
            // 等待服务器响应
            basicResponsePtr basic_response = waitBasicResponse(request_id); // 出现问题
            return basic_response->ok();
        }
        void closeChannel()
        {
            std::string request_id = UUIDer::uuid();

            closeChannelRequest request;
            request.set_channel_id(_channel_id);
            request.set_request_id(request_id);
            _codec->send(_connection, request);
            waitBasicResponse(request_id);
        }
        bool declareExchange(const std::string &exchange_name,
                             ExchangeType type,
                             bool durable,
                             bool auto_delete,
                             const google::protobuf::Map<std::string, std::string> &args_map)
        {
            std::string request_id = UUIDer::uuid();
            // 构造一个声明交换机的请求对象
            declareExchangeRequest request;
            request.set_channel_id(_channel_id);
            request.set_request_id(request_id);
            request.set_exchange_name(exchange_name);
            request.set_exchange_type(type);
            request.set_durable(durable);
            request.set_auto_delete(auto_delete);
            google::protobuf::Map<std::string, std::string> tmp = args_map;
            request.mutable_args()->swap(tmp);

            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            basicResponsePtr basic_response = waitBasicResponse(request_id);

            return basic_response->ok();
        }
        void deleteExchange(const std::string &exchange_name)
        {
            std::string request_id = UUIDer::uuid();
            deleteExchangeRequest request;
            request.set_channel_id(_channel_id);
            request.set_exchange_name(exchange_name);
            request.set_request_id(request_id);
            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            waitBasicResponse(request_id);
        }

        bool declareQueue(const std::string &queue_name,
                          bool durable,
                          bool exclusive,
                          bool auto_delete,
                          const google::protobuf::Map<std::string, std::string> &args_map)
        {
            std::string request_id = UUIDer::uuid();
            declareQueueRequest request;
            request.set_channel_id(_channel_id);
            request.set_request_id(request_id);
            request.set_queue_name(queue_name);
            request.set_durable(durable);
            request.set_auto_delete(auto_delete);
            request.set_exclusive(exclusive);
            google::protobuf::Map<std::string, std::string> tmp = args_map;
            request.mutable_args()->swap(tmp);

            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            basicResponsePtr basic_response = waitBasicResponse(request_id);

            return basic_response->ok();
        }
        void deleteQueue(const std::string &queue_name)
        {
            std::string request_id = UUIDer::uuid();
            deleteQueueRequest request;
            request.set_channel_id(_channel_id);
            request.set_queue_name(queue_name);
            request.set_request_id(request_id);
            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            waitBasicResponse(request_id);
        }
        bool bind(const std::string &exchange_name, const std::string &queue_name, const std::string &binding_key)
        {
            std::string request_id = UUIDer::uuid();
            queueBindRequest request;
            request.set_channel_id(_channel_id);
            request.set_binding_key(binding_key);
            request.set_exchange_name(exchange_name);
            request.set_queue_name(queue_name);
            request.set_request_id(request_id);
            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            basicResponsePtr response = waitBasicResponse(request_id);
            return response->ok();
        }
        void unBind(const std::string &exchange_name, const std::string &queue_name)
        {
            std::string request_id = UUIDer::uuid();
            queueUnBindRequest request;
            request.set_channel_id(_channel_id);
            request.set_exchange_name(exchange_name);
            request.set_queue_name(queue_name);
            request.set_request_id(request_id);
            // 向服务器发送请求
            _codec->send(_connection, request);
            waitBasicResponse(request_id);
        }

        bool basicPublish(const std::string &exchange_name, const BasicProperties *bp, const std::string &body)
        {
            std::string request_id = UUIDer::uuid();
            basicPublishRequest request;
            request.set_channel_id(_channel_id);
            request.set_exchange_name(exchange_name);
            request.set_request_id(request_id);
            request.set_body(body);
            if (bp != nullptr)
            {
                request.mutable_properties()->set_id(bp->id());
                request.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                request.mutable_properties()->set_routing_key(bp->routing_key());
            }
            // 向服务器发送请求
            _codec->send(_connection, request);
            basicResponsePtr response = waitBasicResponse(request_id);
            return response->ok();
        }

        void basicAck(const std::string &message_id)
        {
            if (_consumer.get() == nullptr)
            {
                DEBUG_LOG("当前信道没有消费者，无法进行消息确认");
                return;
            }
            std::string request_id = UUIDer::uuid();
            basicAckRequest request;
            request.set_channel_id(_channel_id);
            request.set_message_id(message_id);
            request.set_queue_name(_consumer->queue_name);
            request.set_request_id(request_id);

            _codec->send(_connection, request);
            waitBasicResponse(request_id);
        }

        bool basicConsume(const std::string &consumer_tag, const std::string &queue_name, bool auto_ack, const ConsumerCallback &cb)
        {
            if (_consumer.get() != nullptr)
            {
                DEBUG_LOG("当前信道已订阅其他消费者，无法进行多次订阅");
                return false;
            }
            std::string request_id = UUIDer::uuid();
            basicConsumeRequest request;
            request.set_channel_id(_channel_id);
            request.set_request_id(request_id);
            request.set_consumer_tag(consumer_tag);
            request.set_queue_name(queue_name);
            request.set_auto_ack(auto_ack);

            _codec->send(_connection, request);
            basicResponsePtr response = waitBasicResponse(request_id);
            if (response->ok() == false)
            {
                DEBUG_LOG("添加订阅失败");
                return false;
            }
            DEBUG_LOG("添加订阅[%s]成功", consumer_tag.c_str());
            _consumer = std::make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb);
            return true;
        }

        void basicCancel()
        {
            // 当前为消费者客户端 ---- 提供给消费者
            if (_consumer.get() == nullptr)
            {
                DEBUG_LOG("取消订阅时，不存在订阅者信息");
                return;
            }
            std::string request_id = UUIDer::uuid();
            basicCancelRequest request;
            request.set_channel_id(_channel_id);
            request.set_request_id(request_id);
            // 注意取消订阅时，直接通过订阅者中的成员来实现！！！因为取消订阅是一定存在消费者的
            request.set_consumer_tag(_consumer->tag);
            request.set_queue_name(_consumer->queue_name);

            _codec->send(_connection, request);
            waitBasicResponse(request_id);
        }
        // 链接收到基础请求，向_basic_response_map添加响应
        void putBasicResponse(const zhong::basicResponsePtr &response)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_response_map.insert(std::make_pair(response->request_id(), response));
            // 收到一个相同获得一个信号量，唤醒waitResponse中的阻塞
            _cv.notify_all();
        }
        // 收到创建用户请求
        void putBasicUserResponse(const basicUserResponsePtr &response)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_user_response_map.insert(std::make_pair(response->request_id(), response));
            // 收到一个相同获得一个信号量，唤醒waitResponse中的阻塞
            _cv.notify_all();
        }
        // 消息处理函数---提供给消费者
        void consume(const zhong::basicConsumeResponsePtr &response)
        {
            // 当前为消费者客户端
            if (_consumer.get() == nullptr)
            {
                DEBUG_LOG("消息处理时，不存在订阅者信息");
                return;
            }
            if (_consumer->tag != response->consumer_tag())
            {
                DEBUG_LOG("消息处理时，响应中订阅者消息[%s]和信道订阅者消息[%s]不同", response->consumer_tag().c_str(), _consumer->tag.c_str());
                return;
            }
            // 调用消费者回调函数进行消费处理
            _consumer->callback(response->consumer_tag(), response->mutable_properties(), response->body());
        }

        std::string channel_id()
        {
            return _channel_id;
        }

        bool declareVirtualHost(const std::string &host_name, bool durable)
        {
            std::string request_id = UUIDer::uuid();
            declareHostRequest request;
            request.set_request_id(request_id);
            request.set_channel_id(_channel_id);
            request.set_host_name(host_name);
            request.set_durable(durable);
            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            basicResponsePtr basic_response = waitBasicResponse(request_id);

            return basic_response->ok();
        }
        void deleteVirtualHost(const std::string &host_name)
        {
            std::string request_id = UUIDer::uuid();
            deleteHostRequest request;
            request.set_request_id(request_id);
            request.set_channel_id(_channel_id);
            request.set_host_name(host_name);
            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            waitBasicResponse(request_id);
        }

        bool selectVirtualHost(const std::string &host_name)
        {
            std::string request_id = UUIDer::uuid();
            selectHostRequest request;
            request.set_request_id(request_id);
            request.set_channel_id(_channel_id);
            request.set_host_name(host_name);
            // 向服务器发送请求
            _codec->send(_connection, request);

            // 等待服务器响应
            basicResponsePtr basic_response = waitBasicResponse(request_id);
            return basic_response->ok();
        }

        bool declareUser(const std::string &user_name, const std::string &password, bool durable, int level)
        {
            std::string request_id = UUIDer::uuid();
            declareUserRequest request;
            request.set_request_id(request_id);
            request.set_channel_id(_channel_id);
            request.set_user_name(user_name);
            request.set_password(password);
            request.set_durable(durable);
            // level我们枚举为 1,2,3
            assert(level <= 3);
            request.set_level(level);
            _codec->send(_connection, request);

            // 等待服务器响应
            basicResponsePtr basic_response = waitBasicResponse(request_id);
            return basic_response->ok();
        }
        void deleteUser(const std::string &user_name, const std::string &password)
        {
            std::string request_id = UUIDer::uuid();
            deleteUserRequest request;
            request.set_request_id(request_id);
            request.set_channel_id(_channel_id);
            request.set_user_name(user_name);
            request.set_password(password);
            _codec->send(_connection, request);

            // 等待服务器响应
            waitBasicResponse(request_id);
        }
        int selectUser(const std::string &user_name, const std::string &password)
        {
            std::string request_id = UUIDer::uuid();
            selectUserRequest request;
            request.set_request_id(request_id);
            request.set_channel_id(_channel_id);
            request.set_user_name(user_name);
            request.set_password(password);
            _codec->send(_connection, request);
            // 等待服务器响应
            basicUserResponsePtr basic_user_response = waitBasicUserResponse(request_id);
            return basic_user_response->level();
        }

    private:
        basicResponsePtr waitBasicResponse(const std::string &request_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&request_id, this]()
                     { return _basic_response_map.find(request_id) != _basic_response_map.end(); });
            basicResponsePtr basic_response = _basic_response_map[request_id];
            _basic_response_map.erase(request_id);
            return basic_response;
        }
        // 用户响应等待函数
        basicUserResponsePtr waitBasicUserResponse(const std::string &request_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [&request_id, this]()
                     { return _basic_user_response_map.find(request_id) != _basic_user_response_map.end(); });
            basicUserResponsePtr basic_user_response = _basic_user_response_map[request_id];
            _basic_user_response_map.erase(request_id);
            return basic_user_response;
        }

    private:
        std::string _channel_id;
        muduo::net::TcpConnectionPtr _connection;
        ProtobufCodecPtr _codec;
        Consumer::ptr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string, zhong::basicResponsePtr> _basic_response_map;
        std::unordered_map<std::string, zhong::basicUserResponsePtr> _basic_user_response_map;
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        Channel::ptr create(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            Channel::ptr channel = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(channel->channel_id(), channel));
            return channel;
        }
        void remove(const std::string &channel_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(channel_id);
        }
        Channel::ptr get(const std::string &channel_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(channel_id);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
};

#endif